package actor

import (
	"runtime"
	"sync"
	"sync/atomic"
)

type localProcess struct {
	ctx           IContext
	mailbox       chan *messageEnvelope
	dead          int32
	scheduleCount int32
}

func newLocalProcess(mailboxSize int32, scheduleCount int32) *localProcess {
	lp := &localProcess{
		mailbox:       make(chan *messageEnvelope, mailboxSize),
		dead:          1,
		scheduleCount: scheduleCount,
	}
	return lp
}

func (lp *localProcess) mainLoop(startedWg *sync.WaitGroup) {
	startedWg.Done()

	var count int32

loop:
	for {
		select {
		case me := <-lp.mailbox:
			if !lp.ctx.processMessage(me.msg, me.sender) {
				break loop
			}
			recycleMessageEnvelope(me)
		}

		count++
		if count == lp.scheduleCount {
			// 每循环一定次数后调用runtime.Gosched调度下
			// 避免此协程太繁忙导致其他协程饥饿的情况
			count = 0
			runtime.Gosched()
		}
	}

	close(lp.mailbox)

	gProcessManager.Remove(lp.ctx.Self())
}

func (lp *localProcess) IsDead() bool {
	return atomic.LoadInt32(&lp.dead) == 1
}

func (lp *localProcess) Start() {
	if !atomic.CompareAndSwapInt32(&lp.dead, 1, 0) {
		return
	}

	var startedWg sync.WaitGroup
	startedWg.Add(1)
	go lp.mainLoop(&startedWg)
	startedWg.Wait()

	lp.SendMessage(startedMessage, lp.ctx.Self())
}

func (lp *localProcess) Stop() {
	if !atomic.CompareAndSwapInt32(&lp.dead, 0, 1) {
		return
	}

	lp.SendMessage(stopMessage, lp.ctx.Self())
}

func (lp *localProcess) SendMessage(message interface{}, sender *PID) {
	me := produceMessageEnvelope()
	me.msg = message
	me.sender = sender

	select {
	case lp.mailbox <- me:
	default:
		logger.Error("send message to process[%s] fail, mailbox is blocked", lp.ctx)
	}
}
